package com.flink.hbase.elasticsearch2kafka;

import com.flink.hbase.kafka2elasticsearch.UserDocument;
import com.fasterxml.jackson.databind.JsonNode;
import org.elasticsearch.search.SearchHit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.Map;

/**
 * 用户文档读取源
 * 继承 ElasticsearchMultithreadedSource，专门用于读取用户文档数据
 */
public class UserDocumentSource extends ElasticsearchMultithreadedSource<UserDocument> {
    private static final Logger LOG = LoggerFactory.getLogger(UserDocumentSource.class);
    
    private static final DateTimeFormatter[] DATE_FORMATTERS = {
        DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"),
        DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"),
        DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS"),
        DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSX"),
        DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS"),
        DateTimeFormatter.ISO_LOCAL_DATE_TIME
    };
    
    public UserDocumentSource(String[] indices, String query, 
                            int scrollSize, int scrollTimeoutMs, 
                            int readThreads, Map<String, Object> esConfig) {
        super(indices, query, scrollSize, scrollTimeoutMs, readThreads, esConfig);
    }
    
    @Override
    protected UserDocument parseHit(SearchHit hit) throws Exception {
        if (hit == null || hit.getSourceAsMap() == null) {
            return null;
        }
        
        try {
            Map<String, Object> source = hit.getSourceAsMap();
            
            // 使用 JSON 解析
            String sourceJson = getObjectMapper().writeValueAsString(source);
            JsonNode jsonNode = getObjectMapper().readTree(sourceJson);
            
            UserDocument.Builder builder = UserDocument.builder();
            
            // 优先使用文档ID作为用户ID，如果没有则从source中获取
            String userId = hit.getId();
            if (userId == null || userId.isEmpty()) {
                userId = getStringValue(jsonNode, "user_id", "userId", "id");
            }
            builder.userId(userId);
            
            // 解析其他字段
            builder.name(getStringValue(jsonNode, "name", "username", "user_name"));
            builder.age(getIntegerValue(jsonNode, "age"));
            builder.gender(getStringValue(jsonNode, "gender", "sex"));
            builder.email(getStringValue(jsonNode, "email", "email_address"));
            builder.phone(getStringValue(jsonNode, "phone", "phone_number", "mobile"));
            builder.address(getStringValue(jsonNode, "address"));
            builder.city(getStringValue(jsonNode, "city"));
            builder.country(getStringValue(jsonNode, "country"));
            builder.status(getStringValue(jsonNode, "status", "user_status"));
            builder.tags(getStringValue(jsonNode, "tags", "user_tags"));
            builder.score(getDoubleValue(jsonNode, "score", "user_score"));
            
            // 解析时间字段
            builder.registrationDate(getDateTimeValue(jsonNode, "registration_date", "reg_date", "created_date"));
            builder.lastLogin(getDateTimeValue(jsonNode, "last_login", "last_login_time"));
            builder.createTime(getDateTimeValue(jsonNode, "create_time", "created_at"));
            builder.updateTime(getDateTimeValue(jsonNode, "update_time", "updated_at"));
            
            UserDocument document = builder.build();
            
            // 设置默认时间
            if (document.getCreateTime() == null) {
                document.setCreateTime(LocalDateTime.now());
            }
            if (document.getUpdateTime() == null) {
                document.setUpdateTime(LocalDateTime.now());
            }
            
            // 验证文档
            if (!document.isValid()) {
                LOG.warn("Invalid document from ES hit {}: missing required fields", hit.getId());
                return null;
            }
            
            return document;
            
        } catch (Exception e) {
            LOG.error("Error parsing ES hit {}", hit.getId(), e);
            return null;
        }
    }
    
    /**
     * 从 JSON 节点获取字符串值，支持多个字段名称
     */
    private String getStringValue(JsonNode node, String... fieldNames) {
        for (String fieldName : fieldNames) {
            JsonNode fieldNode = node.get(fieldName);
            if (fieldNode != null && !fieldNode.isNull()) {
                String value = fieldNode.asText();
                if (value != null && !value.trim().isEmpty()) {
                    return value.trim();
                }
            }
        }
        return null;
    }
    
    /**
     * 从 JSON 节点获取整数值
     */
    private Integer getIntegerValue(JsonNode node, String... fieldNames) {
        for (String fieldName : fieldNames) {
            JsonNode fieldNode = node.get(fieldName);
            if (fieldNode != null && !fieldNode.isNull()) {
                if (fieldNode.isInt()) {
                    return fieldNode.asInt();
                } else if (fieldNode.isTextual()) {
                    try {
                        return Integer.parseInt(fieldNode.asText());
                    } catch (NumberFormatException e) {
                        LOG.warn("Invalid integer value for field {}: {}", fieldName, fieldNode.asText());
                    }
                }
            }
        }
        return null;
    }
    
    /**
     * 从 JSON 节点获取双精度值
     */
    private Double getDoubleValue(JsonNode node, String... fieldNames) {
        for (String fieldName : fieldNames) {
            JsonNode fieldNode = node.get(fieldName);
            if (fieldNode != null && !fieldNode.isNull()) {
                if (fieldNode.isNumber()) {
                    return fieldNode.asDouble();
                } else if (fieldNode.isTextual()) {
                    try {
                        return Double.parseDouble(fieldNode.asText());
                    } catch (NumberFormatException e) {
                        LOG.warn("Invalid double value for field {}: {}", fieldName, fieldNode.asText());
                    }
                }
            }
        }
        return null;
    }
    
    /**
     * 从 JSON 节点获取日期时间值
     */
    private LocalDateTime getDateTimeValue(JsonNode node, String... fieldNames) {
        for (String fieldName : fieldNames) {
            JsonNode fieldNode = node.get(fieldName);
            if (fieldNode != null && !fieldNode.isNull()) {
                String dateTimeStr = fieldNode.asText();
                if (dateTimeStr != null && !dateTimeStr.trim().isEmpty()) {
                    return parseDateTime(dateTimeStr.trim());
                }
            }
        }
        return null;
    }
    
    /**
     * 解析日期时间字符串，支持多种格式
     */
    private LocalDateTime parseDateTime(String dateTimeStr) {
        for (DateTimeFormatter formatter : DATE_FORMATTERS) {
            try {
                return LocalDateTime.parse(dateTimeStr, formatter);
            } catch (DateTimeParseException e) {
                // 继续尝试下一个格式
            }
        }
        
        // 尝试解析时间戳
        try {
            long timestamp = Long.parseLong(dateTimeStr);
            // 判断是秒还是毫秒
            if (timestamp > 1000000000000L) {
                // 毫秒时间戳
                return LocalDateTime.ofEpochSecond(timestamp / 1000, 0, java.time.ZoneOffset.UTC);
            } else {
                // 秒时间戳
                return LocalDateTime.ofEpochSecond(timestamp, 0, java.time.ZoneOffset.UTC);
            }
        } catch (NumberFormatException e) {
            LOG.warn("Unable to parse datetime string: {}", dateTimeStr);
        }
        
        return null;
    }
    
    /**
     * 工厂方法创建 UserDocumentSource 实例
     */
    public static UserDocumentSource create(String[] indices, String query, 
                                          int scrollSize, int scrollTimeoutMs, 
                                          int readThreads, Map<String, Object> esConfig) {
        return new UserDocumentSource(indices, query, scrollSize, scrollTimeoutMs, readThreads, esConfig);
    }
} 